package com.bossien.flink.config;

import com.alibaba.fastjson.JSONObject;
import com.bossien.common.model.student.ExamQuestionAnswer;
import com.bossien.flink.common.PoJoTupleMapFunction;
import com.bossien.flink.common.StreamAndSetEnum;
import com.bossien.flink.interfaces.IExecutionHandlerConfig;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.table.api.StreamQueryConfig;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.TableEnvironment;
import org.apache.flink.table.api.java.StreamTableEnvironment;
import org.springframework.beans.factory.annotation.Autowired;

/**
 * @author 胡春林
 * TableApi基础环境
 * @param <T>
 */
@Slf4j
@Data
public abstract class ExecutionTableConfig<T> implements IExecutionHandlerConfig<T> {

    @Autowired
    RMQExcutionStreamHandlerConfig.RMQExcutionStreamHandlerConfigImpl rmqExcutionStreamHandlerConfig;

    private ExecutionSetHandlerConfig executionSetHandlerConfig = null;

    private  StreamTableEnvironment tableEnvironment;

    private int useStream = 0;

    @Override
    public Table addSource() throws Exception {

       if(useStream == StreamAndSetEnum.DATA_STREAM.getCode())
       {
           DataStream<String> dataStream = rmqExcutionStreamHandlerConfig.addSource();
           DataStream<ExamQuestionAnswer> tuple2DataStream = dataStream.map(new ExamMap());
           tableEnvironment =  TableEnvironment.getTableEnvironment(rmqExcutionStreamHandlerConfig.getStreamExecutionEnvironment());
           //获取table环境配置(具体JOB里面配置)
           Table table = tableEnvironment.fromDataStream(tuple2DataStream);
           tableEnvironment.registerTable(ExamQuestionAnswer.class.getSimpleName(),table);
           //注册table表
           //tableEnvironment.registerDataStream(ExamQuestionAnswer.class.getSimpleName(),tuple2DataStream,"UserId,Score");

          return table;

       }else if(useStream == StreamAndSetEnum.DATA_SET.getCode())
       {
           executionSetHandlerConfig = ExecutionSetHandlerConfig.class.newInstance();
           //to do
           return null;
       }else
       {
           throw  new Exception("usestream参数设置出错,请确认你的配置/mq数据来源对象实例化出错");
       }
    }

    @Override
    public  abstract  T executeFilter(T t);

    @Override
    public abstract T addSkin(T tClass);

    public class ExamMap implements MapFunction<String,ExamQuestionAnswer> {
        @Override
        public ExamQuestionAnswer map(String s) {
            ExamQuestionAnswer examQuestionAnswer = JSONObject.parseObject(s, ExamQuestionAnswer.class);
            return examQuestionAnswer;
        }
    }
}
